package com.caius.xiaohashu.user.relation.biz.service.Impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import com.caius.framework.biz.context.holder.LoginUserContextHolder;
import com.caius.framework.common.exception.BizException;
import com.caius.framework.common.response.PageResponse;
import com.caius.framework.common.response.Response;
import com.caius.framework.common.util.DateUtils;
import com.caius.framework.common.util.JsonUtils;
import com.caius.xiaohashu.user.dto.resp.FindUserByIdRspDTO;
import com.caius.xiaohashu.user.relation.biz.constant.MQConstants;
import com.caius.xiaohashu.user.relation.biz.constant.RedisKeyConstants;
import com.caius.xiaohashu.user.relation.biz.domain.dataobject.FansDO;
import com.caius.xiaohashu.user.relation.biz.domain.dataobject.FollowingDO;
import com.caius.xiaohashu.user.relation.biz.domain.mapper.FansDOMapper;
import com.caius.xiaohashu.user.relation.biz.domain.mapper.FollowingDOMapper;
import com.caius.xiaohashu.user.relation.biz.enums.LuaResultEnum;
import com.caius.xiaohashu.user.relation.biz.enums.ResponseCodeEnum;
import com.caius.xiaohashu.user.relation.biz.model.dto.FollowUserMqDTO;
import com.caius.xiaohashu.user.relation.biz.model.dto.UnfollowUserMqDTO;
import com.caius.xiaohashu.user.relation.biz.model.vo.*;
import com.caius.xiaohashu.user.relation.biz.rcp.UserRcpService;
import com.caius.xiaohashu.user.relation.biz.service.RelationService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Set;

/**
 * @author Caius
 * @description
 * @since Created in 2025-05-22
 */
@Service
@Slf4j
public class RelationServiceImpl implements RelationService {

    @Resource
    private UserRcpService userRcpService;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private FollowingDOMapper followingDOMapper;

    @Resource
    private FansDOMapper fansDOMapper;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    /**
     * 用户关注
     * @param followUserReqVO
     * @return
     */
    @Override
    public Response<?> follow(FollowUserReqVO followUserReqVO) {
        Long followUserId = followUserReqVO.getFollowUserId();

        Long userId = LoginUserContextHolder.getUserId();

        if (Objects.equals(userId, followUserId)) {
            throw new BizException(ResponseCodeEnum.CANT_FOLLOW_YOUR_SELF);
        }

        // 校验用户是否存在
        FindUserByIdRspDTO findUserByIdRspDTO = userRcpService.findById(followUserId);
        if (Objects.isNull(findUserByIdRspDTO)) {
            throw new BizException(ResponseCodeEnum.FOLLOW_USER_NOT_EXISTED);
        }

        String followingRedisKey = RedisKeyConstants.buildUserFollowingKey(followUserId);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_check_and_add.lua")));

        script.setResultType(Long.class);

        LocalDateTime now = LocalDateTime.now();
        // 当前时间转时间戳
        long timestamp = DateUtils.localDateTime2Timestamp(now);
        // 执行 lua 脚本
        Long result = redisTemplate.execute(script, Collections.singletonList(followingRedisKey), followUserId, timestamp);

        LuaResultEnum luaResultEnum = LuaResultEnum.valueOf(result);

        if(Objects.isNull(luaResultEnum)) throw new RuntimeException("Lua 返回结果错误");

        switch (luaResultEnum){
            // 关注列表到达上限
            case FOLLOW_LIMIT -> throw new BizException(ResponseCodeEnum.FOLLOWING_COUNT_LIMIT);
            // 关注列表不存在
            case ALREADY_FOLLOWED -> throw new BizException(ResponseCodeEnum.ALREADY_FOLLOWED);
            // ZSet 关注列表不存在
            case ZSET_NOT_EXIST -> {
                List<FollowingDO> followingDOS = followingDOMapper.selectByUserId(userId);

                long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

                if (CollUtil.isEmpty(followingDOS)) {
                    // 若记录为空，直接 ZADD 关系数据, 并设置过期时间
                    DefaultRedisScript<Long> script1 = new DefaultRedisScript<>();
                    script1.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_add_and_expire.lua")));
                    script1.setResultType(Long.class);

                    redisTemplate.execute(script1, Collections.singletonList(followingRedisKey), followUserId, timestamp, expireSeconds);
                } else {
                    Object[] luaArgs = buildLuaArgs(followingDOS, expireSeconds);

                    // 执行 Lua 脚本，批量同步关注关系数据到 Redis 中
                    DefaultRedisScript<Long> script2 = new DefaultRedisScript<>();
                    script2.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_batch_add_and_expire.lua")));
                    script2.setResultType(Long.class);
                    redisTemplate.execute(script2, Collections.singletonList(followingRedisKey), luaArgs);

                    // 再次调用上面的 Lua 脚本：follow_check_and_add.lua , 将最新的关注关系添加进去
                    result = redisTemplate.execute(script, Collections.singletonList(followingRedisKey), followUserId, timestamp);

                    checkLuaScriptResult(result);
                }
            }
        }

        // 构建消息发送 DTO
        FollowUserMqDTO followUserMqDTO = FollowUserMqDTO.builder()
                .userId(userId)
                .followUserId(followUserId)
                .createTime(now)
                .build();

        // 构建消息对象传递 dto 的 json 数据
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(followUserMqDTO)).build();

        String destination = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW + ":" + MQConstants.TAG_FOLLOW;

        log.info("==> 开始发送关注操作 MQ, 消息体 ：{}", destination);

        String hashKey = String.valueOf(userId);

        rocketMQTemplate.asyncSendOrderly(destination, message, hashKey, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> MQ 发送成功: {}" , sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("==> MQ 发送异常: {}", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 取关用户
     * @param unfollowUserReqVO
     * @return
     */
    @Override
    public Response<?> unFollow(UnfollowUserReqVO unfollowUserReqVO) {
        Long unfollowUserId = unfollowUserReqVO.getUnfollowUserId();
        Long userId = LoginUserContextHolder.getUserId();

        if (Objects.equals(unfollowUserId, userId)) {
            throw new BizException(ResponseCodeEnum.CANT_UNFOLLOW_YOUR_SELF);
        }

        FindUserByIdRspDTO findUserByIdRspDTO = userRcpService.findById(unfollowUserId);

        if (Objects.isNull(findUserByIdRspDTO)) {
            throw new BizException(ResponseCodeEnum.FOLLOW_USER_NOT_EXISTED);
        }

        // 必须是关注用户才能进行取关操作
        String followingRedisKey = RedisKeyConstants.buildUserFollowingKey(unfollowUserId);
        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/unfollow_check_and_delete.lua")));
        script.setResultType(Long.class);

        Long result = redisTemplate.execute(script, Collections.singletonList(followingRedisKey), unfollowUserId);

        if (Objects.equals(result, LuaResultEnum.NOT_FOLLOWED.getCode())) {
            List<FollowingDO> followingDOS = followingDOMapper.selectByUserId(userId);

            long expireSeconds = 60*60*24 + RandomUtil.randomInt(60*60*24);

            if (CollUtil.isEmpty(followingDOS)) {
                throw new BizException(ResponseCodeEnum.NOT_FOLLOWED);
            } else {
                Object[] luaArgs = buildLuaArgs(followingDOS, expireSeconds);

                DefaultRedisScript<Long> script1 = new DefaultRedisScript<>();
                script1.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/follow_batch_add_and_expire.lua")));
                script1.setResultType(Long.class);

                redisTemplate.execute(script1, Collections.singletonList(followingRedisKey), unfollowUserId);

                // 再次调用上面的 Lua 脚本：unfollow_check_and_delete.lua , 将取关的用户删除
                result = redisTemplate.execute(script, Collections.singletonList(followingRedisKey), unfollowUserId);

                if (Objects.equals(result, LuaResultEnum.NOT_FOLLOWED.getCode())) {
                    throw new BizException(ResponseCodeEnum.NOT_FOLLOWED);
                }
            }
        }

        // 发送 MQ
        // 构建消息体 DTO
        UnfollowUserMqDTO unfollowUserMqDTO = UnfollowUserMqDTO.builder()
                .userId(userId)
                .unfollowUserId(unfollowUserId)
                .createTime(LocalDateTime.now())
                .build();

        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(unfollowUserMqDTO)).build();
        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW + ":" + MQConstants.TAG_UNFOLLOW;

        log.info("==> 开始发送取关操作 MQ, 消息体: {}", unfollowUserMqDTO);

        String hashKey = String.valueOf(userId);

        // 发送 MQ 取关操作信息
        rocketMQTemplate.asyncSendOrderly(destination, message, hashKey, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> MQ 发送成功， sendResult: {}" , sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> MQ 发送异常， throwable: {}", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 批量查询用户数据
     * @param findFollowingListReqVO
     * @return
     */
    @Override
    public PageResponse<FindFollowingUserRspVO> findFollowingList(FindFollowingListReqVO findFollowingListReqVO) {
        Long userId = findFollowingListReqVO.getUserId();
        Integer pageNo = findFollowingListReqVO.getPageNo();

        // 从 redis 中间查找
        String followingRedisKey = RedisKeyConstants.buildUserFollowingKey(userId);

        Long total = redisTemplate.opsForZSet().zCard(followingRedisKey);

        List<FindFollowingUserRspVO> findFollowingUserRspVOS = null;

        long limit = 10;

        if (total > 0) {
            long totalPages = PageResponse.getTotalPages(total, limit);

            // 请求页码超出总数
            if(pageNo > totalPages) return PageResponse.success(null, pageNo, total);

            // 准备从 Redis 中查询 ZSet 分页数据
            // 每页 10 个元素，计算偏移量
            long offset = PageResponse.getOffset(pageNo, limit);

            // 使用 ZREVRANGEBYSCORE 命令按 score 降序获取元素，同时使用 LIMIT 子句实现分页
            // 注意：这里使用了 Double.POSITIVE_INFINITY 和 Double.NEGATIVE_INFINITY 作为分数范围
            // 因为关注列表最多有 1000 个元素，这样可以确保获取到所有的元素
            Set<Object> followingUserIdSet = redisTemplate.opsForZSet()
                    .reverseRangeByScore(followingRedisKey, Double.NEGATIVE_INFINITY, Double.POSITIVE_INFINITY, offset, limit);


            if (CollUtil.isNotEmpty(followingUserIdSet)) {
                List<Long> userIds = followingUserIdSet.stream().map(object -> Long.valueOf(object.toString())).toList();

                // RCP 方法
                findFollowingUserRspVOS = rpcUserServiceAndDTO2VO(userIds, findFollowingUserRspVOS);
            }
        } else {
            //  若 Redis 中没有数据，则从数据库查询
            long count = followingDOMapper.selectCountByUserId(userId);
            long totalPage = PageResponse.getTotalPages(count, limit);

            // 请求页码超过总页数
            if(pageNo > totalPage) return PageResponse.success(null, pageNo, count);

            long offset = PageResponse.getOffset(pageNo, limit);

            List<FollowingDO> followingDOS = followingDOMapper.selectPageListByUserId(userId, offset, limit);
            // 赋值真实记录总数
            total = count;

            if (CollUtil.isNotEmpty(followingDOS)) {
                List<Long> userIds = followingDOS.stream().map(FollowingDO::getId).toList();

                // RCP 批量查询用户信息
                rpcUserServiceAndDTO2VO(userIds, findFollowingUserRspVOS);

                // 异步将关注列表全量同步到 Redis
                threadPoolTaskExecutor.submit(() -> syncFollowingList2Redis(userId));
            }
        }
        return PageResponse.success(findFollowingUserRspVOS, pageNo, total);
    }

    /**
     * 批量查找粉丝数据
     * @param findFansListReqVO
     * @return
     */
    @Override
    public PageResponse<FindFansListRspVO> findFansList(FindFansListReqVO findFansListReqVO) {
        Long userId = findFansListReqVO.getUserId();
        Integer pageNo = findFansListReqVO.getPageNo();

        String userFanRedisKey = RedisKeyConstants.buildUserFanKey(userId);

        Long total = redisTemplate.opsForZSet().zCard(userFanRedisKey);
        // 返参
        List<FindFansListRspVO> findFansListRspVOS = null;

        long limit = 10;

        if (total > 0) {
            long totalPages = PageResponse.getTotalPages(total, limit);

            if(pageNo > totalPages) return PageResponse.success(null, pageNo, total);

            long offset = PageResponse.getOffset(pageNo, limit);

            // 使用 ZREVRANGEBYSCORE 命令按 score 降序获取元素，同时使用 LIMIT 子句实现分页
            Set<Object> followingUserIdsSet = redisTemplate.opsForZSet().reverseRangeByScore(userFanRedisKey,
                    Double.NEGATIVE_INFINITY, Double.POSITIVE_INFINITY, offset, limit);

            if (CollUtil.isNotEmpty(followingUserIdsSet)) {
                List<Long> userIds = followingUserIdsSet.stream().map(object -> Long.valueOf(object.toString())).toList();

                // RCP 查询用户信息
                findFansListRspVOS = rpcUserServiceAndCountServiceAndDTO2VO(userIds, findFansListRspVOS);
            }
        } else {
            total = fansDOMapper.selectCountByUserId(userId);

            long totalPages = PageResponse.getTotalPages(total, limit);

            if (pageNo > 500 || pageNo > totalPages) return PageResponse.success(null, pageNo, total);

            long offset = PageResponse.getOffset(pageNo, limit);

            List<FansDO> fansDOS = fansDOMapper.selectPageListByUserId(userId, offset, limit);

            if (CollUtil.isNotEmpty(fansDOS)) {
                List<Long> userIds = fansDOS.stream().map(FansDO::getId).toList();
                // RPC: 调用用户服务、计数服务，并将 DTO 转换为 VO
                findFansListRspVOS = rpcUserServiceAndCountServiceAndDTO2VO(userIds, findFansListRspVOS);
                // 异步将粉丝列表同步到 Redis（最多5000条）
                threadPoolTaskExecutor.execute(() -> syncFansList2Redis(userId));
            }

        }

        return PageResponse.success(findFansListRspVOS, pageNo, total);
    }

    /**
     * 粉丝列表同步到 Redis（最多5000条）
     * @param userId
     */
    private void syncFansList2Redis(Long userId) {
        List<FansDO> fansDOS = fansDOMapper.select5000FansByUserId(userId);

        if (CollUtil.isNotEmpty(fansDOS)) {
            String userFanRedisKey = RedisKeyConstants.buildUserFanKey(userId);

            long expiredSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

            Object[] luaArgs = buildFansZSetLuaArgs(fansDOS, expiredSeconds);

            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/follow_batch_add_and_expire.lua")));
            script.setResultType(Long.class);

            // 执行 lua 脚本
            redisTemplate.execute(script, Collections.singletonList(userFanRedisKey), luaArgs);
        }
    }

    /**
     * 粉丝 redis 查询脚本的参数构建
     * @param fansDOS
     * @param expiredSeconds
     * @return
     */
    private static Object[] buildFansZSetLuaArgs(List<FansDO> fansDOS, Long expiredSeconds) {
        int argsLength = fansDOS.size() * 2 + 1;
        Object[] luaArgs = new Object[argsLength];

        int i = 0;
        for (FansDO fansDO : fansDOS) {
            luaArgs[i]  = DateUtils.localDateTime2Timestamp(fansDO.getCreateTime());
            luaArgs[i+1] = expiredSeconds;
            i += 2;
        }
        // 最后一位过期时间
        luaArgs[argsLength - 1] = expiredSeconds;
        return luaArgs;
    }


    /**
     * RPC: 调用用户服务、计数服务，并将 DTO 转换为 VO 粉丝列表
     * @param ids
     * @param findFansListRspVOS
     * @return
     */
    private List<FindFansListRspVO> rpcUserServiceAndCountServiceAndDTO2VO(List<Long> ids, List<FindFansListRspVO> findFansListRspVOS) {
        List<FindUserByIdRspDTO> findUserByIdRspDTOS = userRcpService.findByIds(ids);

        if (CollUtil.isNotEmpty(findUserByIdRspDTOS)) {
            findFansListRspVOS = findUserByIdRspDTOS.stream()
                    .map(dto -> FindFansListRspVO.builder()
                            .userId(dto.getId())
                            .nickName(dto.getNickName())
                            .avatar(dto.getAvatar())
                            .fansTotal(0L)  // TODO: 这块的数据暂无，后续补充
                            .noteTotal(0L)  // TODO: 这块的数据暂无，后续补充
                            .build())
                    .toList();
        }

        return findFansListRspVOS;
    }

    /**
     * 校验 Lua 脚本结果，根据状态码抛出对应的业务异常
     * @param result
     */
    private static void checkLuaScriptResult(Long result) {
        LuaResultEnum luaResultEnum = LuaResultEnum.valueOf(result);

        if (Objects.isNull(luaResultEnum)) throw new RuntimeException("Lua 返回结果错误");

        switch (luaResultEnum){
            case FOLLOW_LIMIT -> throw new BizException(ResponseCodeEnum.FOLLOWING_COUNT_LIMIT);
            case ALREADY_FOLLOWED -> throw new BizException(ResponseCodeEnum.ALREADY_FOLLOWED);
        }
    }

    /**
     * 构建 lua 脚本参数
     * @param followingDOS
     * @param expireSeconds
     * @return
     */
    private static Object[] buildLuaArgs(List<FollowingDO> followingDOS, long expireSeconds) {
        int argsLength = followingDOS.size() * 2 + 1;
        Object[] luaArgs = new Object[argsLength];

        int i = 0;
        for (FollowingDO followingDO : followingDOS) {
             luaArgs[i] = DateUtils.localDateTime2Timestamp(followingDO.getCreateTime());
             luaArgs[i++] = followingDO.getFollowingUserId();

             i += 2;
        }
        luaArgs[argsLength - 1] =expireSeconds;
        return luaArgs;
    }

    /**
     * 调用 RCP 实现转换 DTO 到 VO
     * @param ids
     * @return
     */
    private List<FindFollowingUserRspVO> rpcUserServiceAndDTO2VO(List<Long> ids, List<FindFollowingUserRspVO> findFollowingUserRspVOS) {
        List<FindUserByIdRspDTO> findUserByIdRspDTOS = userRcpService.findByIds(ids);

        // 转换 DTO -> VO
        if (CollUtil.isNotEmpty(findUserByIdRspDTOS)) {
            findFollowingUserRspVOS = findUserByIdRspDTOS.stream()
                    .map(dto -> FindFollowingUserRspVO.builder()
                            .userId(dto.getId())
                            .avatar(dto.getAvatar())
                            .nickname(dto.getNickName())
                            .introduction(dto.getIntroduction())
                            .build())
                    .toList();
        }
        return findFollowingUserRspVOS;
    }

    /**
     * 全量同步关注列表到 redis
     * @param userId
     */
    private void syncFollowingList2Redis(Long userId){
        // 只能全量关注 1000 人
        List<FollowingDO> followingDOS = followingDOMapper.selectAllByUserId(userId);

        if (CollUtil.isNotEmpty(followingDOS)) {
            String followingRedisKey = RedisKeyConstants.buildUserFollowingKey(userId);
            long expiredSeconds = 60*60*24 + RandomUtil.randomInt(60*60*24);

            Object[] luaArgs = buildLuaArgs(followingDOS, expiredSeconds);

            // lua 脚本全量插入数据
            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/follow_batch_add_and_expire.lua")));
            script.setResultType(Long.class);
            redisTemplate.execute(script, Collections.singletonList(followingRedisKey), luaArgs);
        }
    }
}
